package com.zallds.window;

import com.zallds.utils.Utils;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.RichWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;

/**
 *  全窗口函数
 *  RichWindowFunction
 *  每经过3秒钟，求安放在北京西站各个红外测温仪所探测到的所有旅客的平均体温,和窗口的元数据信息
 */
public class FullWindowFunctionDemo2 {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.socketTextStream("192.168.62.37", 8889)
                .flatMap(new Utils.Splitter())
                .keyBy(value -> value.f0)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(3)))
                .apply(new MyRichWindowFunction())
                .print();

        env.execute();
    }

    private static class MyRichWindowFunction extends RichWindowFunction<Tuple2<String, Integer>, String, String, TimeWindow> {

        @Override
        public void apply(String name, TimeWindow window, Iterable<Tuple2<String, Integer>> elements, Collector<String> out) throws Exception {

            int cnt = 0;
            Long sum = 0L;
            for (Tuple2<String, Integer> element : elements) {
                cnt ++;
                sum = sum + element.f1;
            }

            SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            String start = simpleDateFormat.format(window.getStart());
            String end = simpleDateFormat.format(window.getEnd());

            String result = String.format("红外测温仪→ %s, 最近3秒内,测量到的旅客的平均体温是 %s, 窗口的开始时间是→%s,  窗口的结束时间是→%s",name,sum/cnt,start,end);
            out.collect(result);

        }
    }
}
